package com.bkhech.netty.strategy;

import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.bkhech.enums.MsgTypeEnum;
import com.bkhech.netty.rabbitmq.MessagePublisher;
import com.bkhech.netty.utils.JedisPoolUtil;
import com.bkhech.netty.utils.ZKUtil;
import com.bkhech.netty.websocket.ClientChannelGroup;
import com.bkhech.netty.websocket.UserChannelSession;
import com.bkhech.pojo.netty.ChatMsg;
import com.bkhech.pojo.netty.DataContent;
import com.bkhech.pojo.netty.NettyServerNode;
import com.bkhech.utils.JsonUtils;
import com.bkhech.utils.LocalDateUtils;
import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;
import redis.clients.jedis.Jedis;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

@Slf4j
public class MessageHandlerStrategy {
    private static final ChannelGroup channelGroup = ClientChannelGroup.getInstance();

    private static final Map<MsgTypeEnum, Consumer<Map<String, Object>>> functionMap = new HashMap<>(16);

    static {
        try {
            initConsumerMap();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public static void initConsumerMap() {
        functionMap.put(MsgTypeEnum.CONNECT_INIT, MessageHandlerStrategy::handleInitMsg);
        functionMap.put(MsgTypeEnum.WORDS, MessageHandlerStrategy::handleWordsMsg);
        functionMap.put(MsgTypeEnum.IMAGE, MessageHandlerStrategy::handleImageMsg);
        functionMap.put(MsgTypeEnum.VIDEO, MessageHandlerStrategy::handleVideoMsg);
        functionMap.put(MsgTypeEnum.VOICE, MessageHandlerStrategy::handleVoiceMsg);
    }

    public static void handleMsgByType(Map<String, Object> dataMap) {
        DataContent dataContent = (DataContent) dataMap.get("dataContent");
        ChatMsg chatMsg = dataContent.getChatMsg();
        if (StrUtil.isBlank(chatMsg.getMsgId())) {
            // mps 中提供的雪花算法，得到id
            String idStr = IdWorker.getIdStr();
            chatMsg.setMsgId(idStr);
        }
        Integer msgType = chatMsg.getMsgType();
        MsgTypeEnum msgTypeEnum = MsgTypeEnum.getMsgTypeEnum(msgType);
        functionMap.get(msgTypeEnum).accept(dataMap);
    }

    /**
     * 处理第一次或者短线重连的消息
     *
     * @param dataContentMap
     */
    public static void handleInitMsg(Map<String, Object> dataContentMap) {
        DataContent dataContent = (DataContent) dataContentMap.get("dataContent");
        Channel channel = (Channel) dataContentMap.get("channel");
        // 当 websocket 初次 open 的时候，初始化 Channel，把channel和userId关联起来
        String senderId = dataContent.getChatMsg().getSenderId();
        String currentChannelId = channel.id().asLongText();
        UserChannelSession.putMultiChannels(senderId, channel);
        log.info("用户id是：{}, 对应的channelList的长度是：{}", senderId, UserChannelSession.getMultiChannels(senderId).size());
        UserChannelSession.putChannelIdOfUser(currentChannelId, senderId);

        // 客户端传递过来的节点信息
        NettyServerNode minNode = dataContent.getServerNode();
        //初次链接后，对在线人数进行累加
        ZKUtil.incrementOnlineCount(minNode);

        // 获得port和IP，在redis中进行设置关系，以便在客户端断线之后减少在线人数
        // 因为在 ChatHandler.handlerRemoved 触发时，只知道 userId 和 Channel之间的关系，不知道 userId 和 ip:port 的关系，所以此时存储关系
        JedisPoolUtil.executeInJedis(jedis -> jedis.set(senderId, JsonUtils.objectToJson(minNode)));
    }


    /**
     * 处理文字消息类型
     *
     * 集群情况下的消息处理，此处使用 MQ 异步解耦，保存信息到数据库
     *
     * @param dataContentMap
     * @return
     */
    public static void handleWordsMsg(Map<String, Object> dataContentMap) {
        DataContent dataContent = (DataContent) dataContentMap.get("dataContent");
        ChatMsg chatMsg = dataContent.getChatMsg();

        dataContent.setChatTime(LocalDateUtils.format(chatMsg.getChatTime(), LocalDateUtils.DATETIME_PATTERN_2));

        try {
            // 把聊天消息作为 mq 消息进行广播
            MessagePublisher.sendMsgToOtherNettyServer(JsonUtils.objectToJson(dataContent));
            // 把聊天信息作为mq的消息发送给消费者进行消费处理(保存到数据库)
            MessagePublisher.sendMsgToSave(chatMsg);
        } catch (Exception e) {
            log.error("发送聊天消息到rabbitmq失败:{},{}", e, e.getMessage());
            throw new RuntimeException(e);
        }
    }

    /**
     * 处理文字消息类型,
     *  单机情况下的消息处理
     *
     * @param dataContentMap
     * @return
     */
    public static void handleWordsMsgSingle(Map<String, Object> dataContentMap) {
        // 此处异步解耦，保存信息到数据库
        DataContent dataContent = (DataContent) dataContentMap.get("dataContent");
        ChatMsg chatMsg = dataContent.getChatMsg();

        dataContent.setChatTime(LocalDateUtils.format(chatMsg.getChatTime(), LocalDateUtils.DATETIME_PATTERN_2));

        try {
            // 获取消息接收者的所有在线设备
            List<Channel> receiveChannelList = UserChannelSession.getMultiChannels(chatMsg.getReceiverId());
            if (CollectionUtils.isEmpty(receiveChannelList)) {
                // 如果为空则，表示用户断线或者离线状态，消息不需要发送，后续可以存储到数据库
                chatMsg.setIsReceiverOnLine(false);
            } else {
                chatMsg.setIsReceiverOnLine(true);
                if (MsgTypeEnum.getMsgTypeEnum(chatMsg.getMsgType()) == MsgTypeEnum.VOICE) {
                    chatMsg.setIsRead(false);
                }
                // 发送消息给消息接收者
                UserChannelSession.sendToTarget(receiveChannelList, dataContent);
            }

            String currentChannelId = dataContent.getExtend();
            List<Channel> myOtherChannelList = UserChannelSession.getMyOtherChannel(chatMsg.getSenderId(), currentChannelId);
            // 给自己的其它同步设备也发送消息
            UserChannelSession.sendToTarget(myOtherChannelList, dataContent);

            // 把聊天信息作为mq的消息发送给消费者进行消费处理(保存到数据库)
            MessagePublisher.sendMsgToSave(chatMsg);
        } catch (Exception e) {
            log.error("发送聊天消息到rabbitmq失败:{},{}", e, e.getMessage());
            throw new RuntimeException(e);
        }
    }

    /**
     * 图片处理
     * @param dataContentMap
     */
    public static void handleImageMsg(Map<String, Object> dataContentMap) {
        handleWordsMsg(dataContentMap);
    }

    /**
     * 语音处理
     * @param dataMap
     */
    private static void handleVoiceMsg(Map<String, Object> dataMap) {
        DataContent dataContent = (DataContent) dataMap.get("dataContent");
        ChatMsg chatMsg = dataContent.getChatMsg();
        chatMsg.setIsRead(false);
        dataContent.setChatMsg(chatMsg);
        handleWordsMsg(dataMap);
    }

    /**
     * 视频处理
     *
     * @param dataContentMap
     */
    public static void handleVideoMsg(Map<String, Object> dataContentMap) {
        handleWordsMsg(dataContentMap);
    }

}
